package com.zjl.SpringBoot.第21章_流式_响应式_编程.C_Reactive响应式核心;

import org.junit.Test;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;

import java.io.IOException;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicLong;

public class D5_Flux自己创建序列_单线程 {


    //Sink 接收器，水槽，通道
    //Source:数据源    Sink:接收端
    @Test
    public void generateError() {
        Flux<Object> generate = Flux.generate(sink -> {
            for (int i = 0; i < 10; i++) {
                //传递数据；可能会抛出【不受检异常（运行时异常）、受检异常（编译时异常）】
                sink.next("哈哈-" + i);
            }
        });
        generate.log().subscribe();
    }

    /**
     * generate：产生
     */
    @Test
    public void generate() {
        Flux<Object> generate = Flux.generate(
                () -> 0,//初始值  state
                (state, sink) -> {
                    if (state <= 9) {
                        sink.next("哈哈-" + state);
                    } else {
                        sink.complete();//结束
                    }
                    if (state == 7) {
                        sink.error(new RuntimeException("我不喜欢" + state));
                    }
                    return state + 1;
                });
        generate.log().subscribe();


    }

    @Test
    public void generate2() {
        Flux<Object> generate = Flux.generate(
                AtomicLong::new,//初始值  state
                (state, sink) -> {
                    long andIncrement = state.getAndIncrement();//获取并 + 1
                    if (andIncrement <= 9) {
                        sink.next("哈哈-" + andIncrement);
                    } else {
                        sink.complete();//结束
                    }
                    if (andIncrement == 7) {
                        sink.error(new RuntimeException("我不喜欢" + state));
                    }

                    return state;
                });
        generate.log().subscribe();

    }
    @Test
    public void dispose() throws IOException {
//        Flux<Integer> log = Flux.just(1, 2, 3)
//                .map(i -> i + 2)
//                .log();
        Flux<Integer> log = Flux.range(1, 100)
                .delayElements(Duration.ofSeconds(1))
                .map(i -> i + 2)
                .log();

        Disposable subscribe = log.subscribe();
        new Thread(()->{
            try {
                Thread.sleep(5000);
                /**
                 * 流能够取消，是因为有 {@link Disposable#dispose}  继承了这个接口
                 */
                subscribe.dispose();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        System.in.read();

    }
}
